package com.madhukaraphatak.examples.sparktwo

import org.apache.spark.sql.SparkSession


/**
  * Created by zhang_tao6 on 2017-08-18.
  */
object RDD2DataFrameReflection {

  case class Person(name: String, age: Long)

  def main(args: Array[String]): Unit = {


    System.setProperty("hadoop.home.dir", "D:\\hadoop-2.7.3")
    val path = "src/main/resources/people.json"

    val spark = SparkSession.builder.
      master("local")
      .appName("example")
      .getOrCreate()


    import spark.implicits._

    val df = spark.read.json(path)
    df.show()

    df.printSchema()

    df.select("name").show()
    // +-------+
    // |   name|
    // +-------+
    // |Michael|
    // |   Andy|
    // | Justin|
    // +-------+

    // Select everybody, but increment the age by 1
    df.select($"name", $"age" + 1).show()
    // +-------+---------+
    // |   name|(age + 1)|
    // +-------+---------+
    // |Michael|     null|
    // |   Andy|       31|
    // | Justin|       20|
    // +-------+---------+

    // Select people older than 21
    df.filter($"age" > 21).show()
    // +---+----+
    // |age|name|
    // +---+----+
    // | 30|Andy|
    // +---+----+

    // Count people by age
    df.groupBy("age").count().show()
    // +----+-----+
    // | age|count|
    // +----+-----+
    // |  19|    1|
    // |null|    1|
    // |  30|    1|
    // +----+-----+

    // Register the DataFrame as a SQL temporary view
    df.createOrReplaceTempView("people")

    val sqlDF = spark.sql("SELECT * FROM people")
    sqlDF.show()
    // +----+-------+
    // | age|   name|
    // +----+-------+
    // |null|Michael|
    // |  30|   Andy|
    // |  19| Justin|
    // +----+-------+


    val peopleDS = spark.read.json(path).as[Person]
    peopleDS.show()


    // Create an RDD of Person objects from a text file, convert it to a Dataframe
    val peopleDF = spark.sparkContext
      .textFile("examples/src/main/resources/people.txt")
      .map(_.split(","))
      .map(attributes => Person(attributes(0), attributes(1).trim.toInt))
      .toDF()
    // Register the DataFrame as a temporary view
    peopleDF.createOrReplaceTempView("people")

  }
}
